package bigdata;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import twitter4j.internal.org.json.JSONArray;
import twitter4j.internal.org.json.JSONException;
import twitter4j.internal.org.json.JSONObject;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;


/*this bolt updates our SQL data base with the new user data,  
 * it runs a calculation function to receive the weight of the user
 * and collects statistics of the number of visited pages, and the time each user
 * spend in a single page.
 */


	public  class UpdateBolt extends BaseRichBolt {
	    OutputCollector _collector;
	public static	HashMap<String, Date> timeUserOnSamePage = new HashMap<String, Date>();
	public static	HashMap<String, String> currUserSite = new HashMap<String, String>();
	
	public static final int STATUS_PAGE_VISIT = 1;
	public static final int STATUS_SITE_SEARCH = 15;
	
	    @Override
	    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
	      _collector = collector;
	    }

	    @Override
	    public void execute(Tuple tuple) {
	    	
	    	VisitData visit=(VisitData) tuple.getValue(0);
			float newWeight;
			
			///// pages visits
			Integer pagesVisitedByCurrentUserInLastMin = 0;
			if (visit.status==STATUS_PAGE_VISIT) {
				if(!UpdateAvergeThread.userPageVisitInLastMinMap.containsKey(visit.visitId)) 
					UpdateAvergeThread.userPageVisitInLastMinMap.put(visit.visitId,1);
				
				UpdateAvergeThread.pagesVisitedInLastMin++;
				pagesVisitedByCurrentUserInLastMin = UpdateAvergeThread.userPageVisitInLastMinMap.get(visit.visitId)+1;
	    		UpdateAvergeThread.userPageVisitInLastMinMap.put(visit.visitId,pagesVisitedByCurrentUserInLastMin);
			}
			if (pagesVisitedByCurrentUserInLastMin > UpdateAvergeThread.pageVisitPerMinAvg)
    			visit.numOfPageVisitsInGivenTime = pagesVisitedByCurrentUserInLastMin;
    		////////////////////////
			
			///// Handle Search AVG
			Integer currentUserSearchesInLastMin = 0;
	    	if (visit.status==STATUS_SITE_SEARCH) {
	    		if(!UpdateAvergeThread.userSearchInLastMinMap.containsKey(visit.visitId)) 
					UpdateAvergeThread.userSearchInLastMinMap.put(visit.visitId,1);
				
	    		UpdateAvergeThread.searchesInLastMin++;
	    		currentUserSearchesInLastMin = UpdateAvergeThread.userSearchInLastMinMap.get(visit.visitId)+1;
	    		UpdateAvergeThread.userSearchInLastMinMap.put(visit.visitId,currentUserSearchesInLastMin);
	    	}
    		if (currentUserSearchesInLastMin > UpdateAvergeThread.searchesPerMinAvg)
    			visit.numOfSearchesInGivenTime = currentUserSearchesInLastMin;
    		////////////////////////
	    	
	    	if(!SqlDB.checkIfExists(visit.visitId))
	    	{
	    		SqlDB.insertCustomer(visit.visitId);
	    		
	    	}
    		
	    	WeightCalculator.calculate(visit);
    		
    		SqlDB.updateVisitor(visit);

	    	
	    System.out.print("THIS IS TEST::::" +visit.ip + visit.visitId+ visit.lastIsAlive+":::::");
	 

	    }
	    

	    public void declareOutputFields(OutputFieldsDeclarer declarer) {
	      declarer.declare(new Fields("word"));
	    }
	    
	  }



